package cn.jly.bigdata.spark.core

import java.util

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{AccumulatorV2, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/28 22:08
 */
object Spark14_Accumulator {

  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("test-hbase"))

    val dataRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
    val res: Int = dataRdd.reduce(_ + _)
    println(s"res = $res")

    println("-----------------------")
    var sum = 0
    // 每个executor拿到了只是driver中变量sum的副本，对副本的操作并不会影响到driver中sum的值
    // 或者说只是driver把sum变量序列化传给executor,在executor操作完sum后，不会且无法将sum回传给driver
    dataRdd.foreach(sum += _)
    println(s"sum = $sum")

    println("-----------------------")
    // 使用累加器来共享变量，来累加数据
    val accumulator: LongAccumulator = sc.longAccumulator
    dataRdd.foreach {
      case i =>
        // 执行累加器的累加功能
        accumulator.add(i)
    }
    // 获取累加器的值
    println(s"accumulator = ${accumulator.value}")

    println("-----------------------")
    // 使用自定义的累加器
    val wordAccumulator = new WordAccumulator
    // 先向spark注册累加器
    sc.register(wordAccumulator)

    val dataRdd2: RDD[String] = sc.makeRDD(List("hadoop", "scala", "hive", "zookeeper", "spark"))
    // todo 执行累加器的添加功能
    dataRdd2.foreach(wordAccumulator.add)
    // todo 获取累加器的值
    println(wordAccumulator.value)


    sc.stop()
  }
}

// 自定义累加器
class WordAccumulator extends AccumulatorV2[String, util.ArrayList[String]] {

  private val _wordList = new util.ArrayList[String]()

  /**
   * 当前累加器是否为初始化状态
   *
   * @return
   */
  override def isZero: Boolean = {
    _wordList.isEmpty
  }

  /**
   * 复制累加器
   *
   * @return
   */
  override def copy(): AccumulatorV2[String, util.ArrayList[String]] = {
    val newAcc = new WordAccumulator
    _wordList.synchronized {
      newAcc._wordList.addAll(this._wordList)
    }

    newAcc
  }

  /**
   * 重置累加器
   */
  override def reset(): Unit = {
    _wordList.clear()
  }

  /**
   * 添加
   *
   * @param v
   */
  override def add(v: String): Unit = {

    // 如果单词包含“p”,则添加进来
    if (v.contains("p")) {
      _wordList.add(v)
    }
  }

  /**
   * 不同executor的两个累加器合并
   *
   * @param other
   */
  override def merge(other: AccumulatorV2[String, util.ArrayList[String]]): Unit = {

    // 如果是累加器对象，则合并集合元素
    other match {
      case wordAccumulator: WordAccumulator => _wordList.addAll(wordAccumulator.value)
    }
  }

  override def value: util.ArrayList[String] = _wordList
}